package org.ricks.ioc.scheduler;

import org.ricks.ioc.event.DelayEvent;
import org.ricks.ioc.utils.Logger;
import org.ricks.ioc.utils.StringUtils;
import org.ricks.ioc.utils.ThreadUtils;

import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.ReentrantLock;
import java.util.function.Function;

/**
 * zfoo 只支持cron
 * 1.数据结构优化：采用优先队列（PriorityQueue）替代CopyOnWriteArrayList，配合ReentrantLock实现线程安全，时间复杂度从O(n)降为O(log n)
 * 2.异步执行机制：引入CachedThreadPool处理任务执行，避免阻塞调度线程
 * 3.时间回拨处理：优化时间校准逻辑，仅重新计算受影响任务
 * 4.线程安全优化：使用双重锁机制保证队列操作的原子性
 * 5.异常处理：增加异步任务执行的异常捕获机制
 */
public abstract class SchedulerBus {

    private static volatile boolean isShutdown = false;
    // 任务存储及映射
    private static final ConcurrentHashMap<String, SchedulerDefinition> taskMap = new ConcurrentHashMap<>();

    private static final ConcurrentSkipListSet<SchedulerDefinition> schedulerQueue = new ConcurrentSkipListSet<>(
            Comparator.comparingLong(SchedulerDefinition::getTriggerTimestamp).thenComparing(SchedulerDefinition::getTaskId)
    );

    private static final ReentrantLock mainLock = new ReentrantLock();
    private static final int CPU_CORES = Runtime.getRuntime().availableProcessors();
    private static final AtomicInteger pendingTasks = new AtomicInteger(0);

    // 任务执行线程池（支持动态扩容）
    private static final ExecutorService taskExecutor = new ThreadPoolExecutor(
            CPU_CORES * 2,
            CPU_CORES * 8,
            60L, TimeUnit.SECONDS,
            new LinkedBlockingQueue<>(CPU_CORES * 2000),
            new SchedulerThreadFactory(2),
            new ThreadPoolExecutor.CallerRunsPolicy()
    );

    // 调度线程池[8](@ref)
    private static final ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor(
            new SchedulerThreadFactory(1));

    private static long lastTriggerTimestamp = 0L;

    static {
        executor.scheduleAtFixedRate(() -> {
            try {
                triggerPerSecond();
            } catch (Throwable e) {
                Logger.error("Scheduler trigger error", e);
            }
        }, 0, 100, TimeUnit.MILLISECONDS);
    }

    // 线程工厂优化[6](@ref)
    private static class SchedulerThreadFactory implements ThreadFactory {
        private final int poolNumber;
        private final AtomicInteger threadNumber = new AtomicInteger(1);

        SchedulerThreadFactory(int poolNumber) {
            this.poolNumber = poolNumber;
        }

        @Override
        public Thread newThread(Runnable r) {
            Thread thread = new Thread(r, StringUtils.format("scheduler-p{}-t{}", poolNumber, threadNumber.getAndIncrement()));
            thread.setDaemon(false);
            thread.setPriority(Thread.NORM_PRIORITY);
            thread.setUncaughtExceptionHandler((t, e) -> Logger.error(t.toString(), e));
            ThreadUtils.registerSingleThreadExecutor(thread, executor);
            return thread;
        }
    }

    private static void triggerPerSecond() {
        final long currentTime = System.currentTimeMillis();

        mainLock.lock();
        try {
            handleTimeRollback(currentTime);
            processTriggeredTasks(currentTime);
        } finally {
            mainLock.unlock();
        }
    }

    /**
     * 时间回拨处理（优化后）
     */
    private static void handleTimeRollback(long currentTime) {
        if (currentTime >= lastTriggerTimestamp) {
            lastTriggerTimestamp = currentTime;
            return;
        }

        // 使用NavigableSet快速定位受影响任务
        SchedulerDefinition dummy = new SchedulerDefinition(currentTime);
        NavigableSet<SchedulerDefinition> affected = schedulerQueue.tailSet(dummy, true);

        List<SchedulerDefinition> recalibrated = new ArrayList<>(affected.size());
        affected.forEach(task -> {
            recalibrated.add(task);
            pendingTasks.decrementAndGet();
        });
        affected.clear();

        recalibrated.forEach(task -> {
            task.calibrateTime(currentTime);
            if (task.updateNextTriggerTime()) {
                if (schedulerQueue.add(task)) {
                    pendingTasks.incrementAndGet();
                }
            } else {
                taskMap.remove(task.getTaskId());
            }
        });

        lastTriggerTimestamp = currentTime;
    }

    // 异步任务处理[8](@ref)
    private static void processTriggeredTasks(long currentTime) {
        while (true) {
            SchedulerDefinition task = schedulerQueue.pollFirst();
            if (task == null) break;

            pendingTasks.decrementAndGet();

            if (task.isCancelled()) {
                taskMap.remove(task.getTaskId());
                continue;
            }

            if (task.getTriggerTimestamp() > currentTime) {
                schedulerQueue.add(task);
                pendingTasks.incrementAndGet();
                break;
            }

            if (rescheduleTask(task, currentTime)) {
                submitAsyncExecution(task);
            }
        }
    }

    private static boolean rescheduleTask(SchedulerDefinition task, long currentTime) {
        if (!task.updateNextTriggerTime()) {
            taskMap.remove(task.getTaskId());
            return true;
        }
        if (schedulerQueue.add(task)) {
            pendingTasks.incrementAndGet();
        }
        return true;
    }


    private static void submitAsyncExecution(SchedulerDefinition task) {
        taskExecutor.execute(() -> {
            try {
                if (!task.isCancelled()) {
                    task.getScheduler().invoke();
                }
            } catch (Throwable e) {
                Logger.error("Task execution failed: " + task.getTaskId(), e);
                // 自定义重试逻辑
                if (task.checkRetry()) {
                    rescheduleTask(task, System.currentTimeMillis());
                }
            }
        });
    }

    /**
     * 注册调度任务
     */
    public static String registerTask(SchedulerDefinition task) {
        if (isShutdown) throw new IllegalStateException("Scheduler shutdown");
        mainLock.lock();
        try {
            if (taskMap.putIfAbsent(task.getTaskId(), task) != null) {
                throw new IllegalArgumentException("Duplicate task ID: " + task.getTaskId());
            }
            if (schedulerQueue.add(task)) {
                pendingTasks.incrementAndGet();
            }
            return task.getTaskId();
        } finally {
            mainLock.unlock();
        }
    }

    /**
     * 取消任务（原子操作）
     */
    public static boolean cancelTask(String taskId) {
        mainLock.lock();
        try {
            SchedulerDefinition task = taskMap.remove(taskId);
            if (task != null) {
                task.cancel();
                if (schedulerQueue.remove(task)) {
                    pendingTasks.decrementAndGet();
                }
                return true;
            }
            return false;
        } finally {
            mainLock.unlock();
        }
    }


    /**
     * 注册周期性任务
     * @param initialDelay 首次执行延迟(ms)
     * @param period 执行周期(ms)
     * @return taskId 任务唯一标识
     */
    public static String scheduleAtFixedRate(Runnable runnable,long initialDelay, long period) {
        return registerTask(runnable, r -> SchedulerDefinition.createFixJob(initialDelay,period, runnable));
    }

    // 改造schedule方法
    public static String schedule(Runnable runnable, long delay) {
        return registerTask(runnable, r -> SchedulerDefinition.createDelayJob(delay, r));
    }

    public static String schedule(DelayEvent event) {
        SchedulerDefinition schedulerDefinition = SchedulerDefinition.createDelayEventJob(event);
        return registerTask(schedulerDefinition);
    }

    private static String registerTask(Runnable runnable, Function<Runnable, SchedulerDefinition> creator) {
        SchedulerDefinition task = creator.apply(runnable);
        return registerTask(task);
    }

    /**
     * cron表达式执行的任务
     */
    public static String scheduleCron(Runnable runnable, String cron) {
        return registerTask(runnable, r -> SchedulerDefinition.createCronJob(cron, r));
    }

    public static boolean checkTaskId (String taskId) {
        return taskMap.containsKey(taskId);
    }


    // 添加统计方法
    public static int getPendingTaskCount() {
        return pendingTasks.get();
    }

    public static void shutdown() {
        isShutdown = true;
        executor.shutdown();
        taskExecutor.shutdown();
        try {
            if (!executor.awaitTermination(10, TimeUnit.SECONDS)) {
                executor.shutdownNow();
            }
            if (!taskExecutor.awaitTermination(10, TimeUnit.SECONDS)) {
                taskExecutor.shutdownNow();
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }
}
